//
// Created by shen on 2025/2/27.
//

#include "SafeQueue.h"
#include "LogUtil.h"


SafeQueue::~SafeQueue() {
    clearAvPacket();
    condPacket.notify_one();
}

int SafeQueue::putAvPacket(AVPacket *avPacket) {
    std::unique_lock<std::mutex> lock(mutex);
    queuePacket.push(avPacket);
    LOGCATE("size = %d ",queuePacket.size());
    condPacket.notify_all();

    return 0;
}

int SafeQueue::getAvPacket(AVPacket *avPacket) {

    if (ffStatus!= nullptr&&!ffStatus->exit) {
        std::unique_lock<std::mutex> lock(mutex);

        while (!(ffStatus->exit)) {
            if (!queuePacket.empty()) {

                auto *packet1 = queuePacket.front();
                int ret = av_packet_ref(avPacket, packet1);
                if (ret == 0) {
                    queuePacket.pop();
                }

                av_packet_free(&packet1);

                av_free(packet1);

                packet1 = nullptr;

                //LOGE("取出一个packet  当前剩余 %d 个",queuePacket.size());
                break;

            } else {
                if (ffStatus!= nullptr&&!ffStatus->exit) {
//                  condPacket.wait(lock, []{ return condition; });
                    condPacket.wait(lock);
                }
            }
        }

    }
    MESSAGEI("getAvPacket6");
    return 0;
}

int SafeQueue::getQueueSize() {
    std::unique_lock<std::mutex> lock(mutex);
    return queuePacket.size();
}

void SafeQueue::clearAvPacket() {
    condPacket.notify_one();
    std::unique_lock<std::mutex> lock(mutex);
    while(!queuePacket.empty()){
        auto *avPacket=queuePacket.front();
        queuePacket.pop();
        av_packet_free(&avPacket);
        av_free(avPacket);
        avPacket= nullptr;
    }

}

void SafeQueue::noticeQueue() {
    condPacket.notify_one();
}

double SafeQueue::getMaxPts() {
    std::unique_lock<std::mutex> lock(mutex);
    double pts=0;
    while (!queuePacket.empty()){
        auto  * avPacket=queuePacket.front();
        if (avPacket->pts>pts){
            pts=avPacket->pts;
        }
    }
    return pts;
}

void SafeQueue::clearByBeforeTime(int64_t time, AVRational time_base) {
    std::unique_lock<std::mutex> lock(mutex);

    while(queuePacket.size()>50){
        auto *avPacket=queuePacket.front();
        if (avPacket->pts* av_q2d(time_base)<time){
            queuePacket.pop();
            av_packet_free(&avPacket);
            av_free(avPacket);
            avPacket = NULL;
        }else{
            break;
        }
    }
}

SafeQueue::SafeQueue(FFStatus *ffStatus) {
    this->ffStatus=ffStatus;
}
